package com.example.consumer.service;

import com.example.consumer.config.RedisStreamConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

/**
 * @Author: AdobePeng
 * @Date: 2021/3/12 15:35
 */
@Component
public class OrderStreamListener implements StreamListener<String, ObjectRecord<String, String>> {
    static final Logger LOGGER = LoggerFactory.getLogger(OrderStreamListener.class);
    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Autowired
    RedisStreamConfig redisStreamConfig;

    @Override
    public void onMessage(ObjectRecord<String, String> message) {
        try{
            // 消息ID
            RecordId messageId = message.getId();

            // 消息的key和value
            String string = message.getValue();
            LOGGER.info("StreamMessageListener  stream message。messageId={}, stream={}, body={}", messageId, message.getStream(), string);
            // 通过RedisTemplate手动确认消息
            this.stringRedisTemplate.opsForStream().acknowledge(redisStreamConfig.getGroup(), message);
        }catch (Exception e){
            // 处理异常
            e.printStackTrace();
        }

    }
}